import ssl
from kafka import KafkaProducer, KafkaConsumer
from kafka.errors import kafka_errors
import traceback
import json


context = ssl.create_default_context()
context.options &= ssl.OP_NO_TLSv1
context.options &= ssl.OP_NO_TLSv1_1
# context.load_verify_locations(cafile=/secure/cacerts)
context.check_hostname = False


def producer_demo():
    # 假设生产的消息为键值对（不是一定要键值对），且序列化方式为json
    producer = KafkaProducer(
        bootstrap_servers=['localhost:9092'], 
        key_serializer=lambda k: json.dumps(k).encode(),
        value_serializer=lambda v: json.dumps(v).encode(),
        # security_protocol='SSL', ssl_context=context,
        # api_version=(0, 10, 2)
        # api_version=(2, 8, 1)
        )
    # 发送三条消息
    for i in range(0, 3):
        future = producer.send(
            'test_input',
            key='count_num',  # 同一个key值，会被送至同一个分区
            value=str(i),
            partition=0)  # 向分区1发送消息
        print("send {}".format(str(i)))
        try:
            future.get(timeout=3) # 监控是否发送成功           
        except Exception as e:  # 发送失败抛出kafka_errors
            print(str(e))
            traceback.format_exc()

producer_demo()